package com.ruoyi.flink.table;

import com.ruoyi.flink.bean.TaxiFare;
import com.ruoyi.flink.source.MyTaxiFareSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @program: ruoyi
 * @description: BlinkStreamingTable
 * @author: zengchen
 * @create: 2021-03-05 19:35
 **/
public class BlinkStreamingTable {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);

//        bsTableEnv.useCatalog("taxi_fare_catalog");
//        bsTableEnv.useCatalog("taxi_fare_database");

        DataStreamSource<TaxiFare> taxiFareDataStreamSource = bsEnv.addSource(new MyTaxiFareSource());
        bsTableEnv.createTemporaryView("b20210306",taxiFareDataStreamSource);
        Table taxiFares = bsTableEnv.from("b20210306");
        Table result = taxiFares
                .groupBy($("rideId"))
                .select($("rideId"), $("totalFare").sum().as("sumFare"));
//                .orderBy($("sumFare").desc());
//        taxiFares.tem

        System.out.println(result.explain());


        bsTableEnv.executeSql("select * from result")
//                .toAppendStream(result, Row.class)
                .print();

        bsEnv.execute("BlinkStreamingTable");
    }
}
